package com.lagou.mall.esflow.server.domain;

import static com.lagou.mall.esflow.client.AckEnum.ACK;
import static com.lagou.mall.esflow.server.constant.ESFlowConstant.ITEM_DOCS;

import com.lagou.mall.esflow.client.AckEnum;
import com.lagou.mall.esflow.client.Binlog;
import com.lagou.mall.esflow.server.config.ElasticSearchTemplate;
import com.lagou.mall.esflow.server.config.EnvConfig;
import com.lagou.mall.esflow.server.domain.index.ItemIndex;
import com.lagou.mall.esflow.server.domain.index.ItemIndex.ItemDoc;
import com.lagou.mall.esflow.server.handler.AbstractTopicConsumerHandler;
import com.lagou.mall.esflow.server.persistence.model.ItemDO;
import com.lagou.mall.esflow.server.repository.ItemRepository;
import com.lagou.mall.esflow.server.util.JsonUtil;
import java.util.Objects;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class ItemIndexConsumerHandler extends AbstractTopicConsumerHandler {

    @Resource
    private EnvConfig envConfig;
    @Resource
    private ElasticSearchTemplate elasticSearchTemplate;
    @Resource
    private ItemRepository itemRepository;


    @Override
    public AckEnum start(KafkaConsumer consumer, Binlog binlog) {
        String tableName = super.getTable(binlog);
        String idValue = super.getId(binlog);
        if ("item".equals(tableName)) {
            return processWithItem(idValue);
        }
        log.info("table {} id {} no processing", tableName, idValue);
        return ACK;
    }

    /**
     * 索引主键全量更新
     */
    private AckEnum processWithItem(String idValue) {
        long itemId = Long.parseLong(idValue);
        ItemDO itemDO = itemRepository.getItem(itemId);
        if (Objects.isNull(itemDO.getId())) {
            log.info("db not itemDO data by id {}", idValue);
            return ACK;
        }
        ItemDoc itemDoc = new ItemDoc(); //此处实际上是吧itemDO转换为itemDoc
        ItemIndex index = ItemIndex.builder()
                .itemDoc(itemDoc)
                .build();
        String body = JsonUtil.toString(index);
        elasticSearchTemplate.upsert(String.valueOf(itemId), super.assembleIndex(envConfig, ITEM_DOCS),
                body);
        return ACK;
    }


}
